package cn.lgwen.candy.runtime.core;

import cn.lgwen.candy.runtime.operator.*;
import org.junit.Test;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 */
public class RunTask {

    @Test
    public void JobGraph() {
        Environment environment = ExecutorEnvironment.getEnvironment();
        Stream<String> source = environment.addSource(new SourceOperator<String>() {
            @Override
            public void run(Fetcher<String> collector) {

            }
        });
        Stream<Integer> stream = source.processor(new Processor<String, Integer>() {
            @Override
            public void process(String input, Collector<Integer> out) {
                out.collect(Integer.valueOf(input));
            }
        });
        stream.addSink(new SinkOperator<Integer>() {
            @Override
            public void invoke(Integer input) {
                System.out.println(input);
            }
        });
        Graph graph = environment.createGraph();
        graph.toString();
    }

    @Test
    public void runTask() throws Exception{
        Environment environment = ExecutorEnvironment.getEnvironment();
        SourceOperator<String> sourceOperator = new SourceOperator<String>() {
            @Override
            public void run(Fetcher<String> collector) {
                collector.produce("1-2-3-4");
            }
        };
        Stream<String> source = environment.addSource(sourceOperator);
        Processor<String, Integer> flatMap = new Processor<String, Integer>() {
            @Override
            public void process(String input, Collector<Integer> out) {
                String[] strings = input.split("-");
                for(String str : strings) {
                    out.collect(Integer.valueOf(str));
                }
            }
        };
        Stream<Integer> stream = source.processor(flatMap);
        SinkOperator<Integer> sinkOperator = new SinkOperator<Integer>() {
            @Override
            public void invoke(Integer input) {
                System.out.println(input);
            }
        };
        stream.addSink(sinkOperator);
        environment.executor();
    }
}
